package com.haoziqi.chapter_02;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Properties;

/**
 * description
 * created by A on 2021/3/4
 */
public class kafkaSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        Properties pro = new Properties();
        pro.setProperty("bootstrap.servers","hadoop102:9092,hadoop103:9092,hadoop104:9092");
        pro.setProperty("group.id", "Flink01_Source_Kafka");
        pro.setProperty("auto.offset.reset", "latest");
        //这里可以点进方法，查看泛型的解释，这里泛型的意思是设置传过来的数据源的类型
        environment.addSource(new FlinkKafkaConsumer<String>("sender",new SimpleStringSchema(),pro)).print();
        environment.execute();
    }
}
